package com.dahua.messagesyncbj.controller;


import com.alibaba.fastjson2.JSON;
import com.dahua.messagesyncbj.pojo.MeasureInfo;
import com.dahua.messagesyncbj.pojo.SyncMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.ExecutionException;

@RestController
@RequestMapping("/kafka")
@Slf4j
public class KafkaController {
    @Resource
    private AdminClient adminClient;
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    /*  外部同步测点信息接口*/
    @PostMapping("/cmode")
    public void cmode(@RequestBody MeasureInfo measureInfo) {
        kafkaTemplate.send("CMODE_DATA", JSON.toJSONString(measureInfo));
    }

    /*  外部同步报警接口*/
    @PostMapping("/sync")
    public void sync(@RequestBody SyncMessage syncMessage) {
        kafkaTemplate.send("SYNC_DATA", JSON.toJSONString(syncMessage));
    }

    @GetMapping("{topic}")
    public String message(@PathVariable("topic") String topic, @RequestBody MeasureInfo msg) {
        String s = JSON.toJSONString(msg);
        kafkaTemplate.send(topic, s);
        return "生产者生产消息： topic = {yf_test}， msg={" + msg + "}";
    }

    @GetMapping("topic/list")
    public String list() throws ExecutionException, InterruptedException {
        ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
        listTopicsOptions.listInternal(true);
        ListTopicsResult result = adminClient.listTopics(listTopicsOptions);
        Collection<TopicListing> list = result.listings().get();
        return list.toString();
    }

    @PostMapping("topic/create")
    public void create(@RequestBody String topics) {
        if (StringUtils.isEmpty(topics)) {
            return;
        }
        Collection<NewTopic> newTopics = new ArrayList<>();
        Arrays.stream(topics.split(",")).forEach(topic -> {
            NewTopic newTopic = new NewTopic(topic, 1, (short) 1);
            newTopics.add(newTopic);
        });
        adminClient.createTopics(newTopics);
    }
}
